package com.leilei.join.common;

import cn.hutool.core.util.RandomUtil;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.time.LocalDate;
import java.time.format.DateTimeFormatter;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/27 16:52
 * @desc 车辆详情数据源
 */
public class VehicleDetailSource implements SourceFunction<VehicleDetail> {
    private Boolean flag = true;
    private String[] colors = new String[]{"黄", "黑", "宝石蓝"};

    @Override
    public void run(SourceContext<VehicleDetail> ctx) throws Exception {
        while (flag) {
            int vehicleId = RandomUtil.randomInt(1, 4);
            VehicleDetail vehicleDetail = VehicleDetail.builder()
                    .id(RandomUtil.randomInt())
                    .vehicleId(vehicleId)
                    .color(colors[vehicleId - 1])
                    .createTime(Integer.parseInt(LocalDate.now().format(DateTimeFormatter.BASIC_ISO_DATE)))
                    .build();
            ctx.collect(vehicleDetail);
            Thread.sleep(2000);
        }
    }

    @Override
    public void cancel() {
        flag = false;
    }
}